package com.shujia.flink.state;

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Demo3ValueState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> wordsDS = env.socketTextStream("master", 8888);

        //分组
        KeyedStream<String, String> keyByDS = wordsDS.keyBy(word -> word);

        /*
         * process算子时flink提供的一个底层算子，可以获取到flink底层的状态，时间和数据
         */
        DataStream<Tuple2<String, Integer>> countDS = keyByDS
                .process(new KeyedProcessFunction<String, String, Tuple2<String, Integer>>() {

                    //ValueState：单值状态，时flink给我们提供的状态开发入口，会被checkpoint定时保存到hdfs中，保证状态不丢失
                    ValueState<Integer> valueState;

                    //open方法每一个task启动的时候执行一次，一般用于初始化
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //获取flink环境对象
                        RuntimeContext context = getRuntimeContext();

                        //创建状态的描述对象。指定状态的类型和名称
                        ValueStateDescriptor<Integer> valueStateDescriptor = new ValueStateDescriptor<>("count", Types.INT);

                        //初始化状态
                        //ValueState： 单值状态，为每一个key在状态中保存一个值
                        valueState = context.getState(valueStateDescriptor);
                    }

                    /**
                     * processElement方法每一条数据执行一次
                     * @param word 一行数据
                     * @param ctx 上下文对象,可以获取到flink的key和时间属性
                     * @param out 用于将处理结果发送到下游
                     */
                    @Override
                    public void processElement(String word,
                                               KeyedProcessFunction<String, String, Tuple2<String, Integer>>.Context ctx,
                                               Collector<Tuple2<String, Integer>> out) throws Exception {
                        //获取状态中保存和值
                        Integer count = valueState.value();
                        if (count == null) {
                            count = 0;
                        }
                        //累加季孙
                        count++;
                        //将计算结果发送到下游
                        out.collect(Tuple2.of(word, count));
                        //更新状态中保存的结果
                        valueState.update(count);
                    }
                });

        countDS.print();

        env.execute();
    }
}
